package com.ibm.cps.publisher;

import java.io.IOException;

import org.apache.log4j.Logger;

import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.messagequeue.AMQPClient;
import com.ibm.cps.newmessage.MetadataInputSchema;
import com.ibm.cps.newmessage.MetadataInputSchema.FieldType;
import com.ibm.interfaces.AbstractMessagePublisher;
import com.ibm.interfaces.IMessageClient;
import com.ibm.interfaces.IMessageParser;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class MessageQueuePublisher extends AbstractMessagePublisher {
	private static final Logger logger = Logger.getLogger(MessageQueuePublisher.class);

	private String server;
	private String username;
	private String password;
	private int port;
	private String topicname;
	private IMessageClient client;

	public MessageQueuePublisher(String metadata) throws CPSException {
		super(metadata);
		ObjectNode fields = JSONUtility.fromObjectString(metadata);
		server = fields.get(MessageFields.HOSTNAME).asText();
		username = fields.get(MessageFields.USERNAME).asText();
		password = fields.get(MessageFields.PASSWORD).asText();
		port = fields.get(MessageFields.PORT).asInt();
		topicname = fields.get(MessageFields.TOPICNAME).asText();
		createTopic();
		client = new AMQPClient(server, port, username, password);
	}
	
	private void createTopic() throws CPSException {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost(server);
		factory.setPort(port);
		factory.setUsername(username);
		factory.setPassword(password);
		try {
			Connection connection = factory.newConnection();
			Channel channel = connection.createChannel();
			channel.exchangeDeclare(topicname, "fanout", false, false, null);
			channel.queueDeclare(topicname, false, false, false, null);
			channel.queueBind(topicname, topicname, topicname);
		}catch(IOException failure){
			logger.error("Failed to create rabbitmq topic " + topicname);
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"Can't create topic " + topicname + " on server " + server + ":" + port + " successfully.");
		}
	}

	@Override
	public void publishMessage(ParsedDataMessage parsedMessage, IMessageParser schema, boolean isBuffer) throws CPSException {

		/*if (parsedMessage.getValueCount() != schema.getTagCount()) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"The count of values( " + Arrays.toString(parsedMessage.getValues())
							+ ") should be equal to schema tag count (" + Arrays.toString(schema.getTagNames()) + ").");
		}*/
		client.publishMessage(topicname, schema.restoreFromDataMessage(parsedMessage));
	}

	@Override
	public MetadataInputSchema getInputSchema() {
		MetadataInputSchema schema = new MetadataInputSchema();
		String[] fields = { MessageFields.DESTINATION, MessageFields.HOSTNAME, MessageFields.NAME,
				MessageFields.PASSWORD, MessageFields.USERNAME, MessageFields.PROTOCOL_TYPE, MessageFields.TOPICNAME };

		schema.addField(FieldType.String, fields);
		schema.addField(FieldType.Integer, MessageFields.PORT);
		return schema;
	}

	@Override
	public void close() {
		client.close();
	}

}
